package com.example.provide;

import com.example.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class Provider {

    //队列名称
    public static final String QUEUE_NAME = "mirror_hello";

    //发送消息到队列中
    public static void sendMS() {
        Connection connection = null;
        Channel channel = null;
        try {
            //创建连接
            connection = RabbitMqUtils.getConnection();
            //获取信道
            channel = connection.createChannel();
            //开启单个发布确认
            channel.confirmSelect();
            /* 创建队列 queueDeclare
             * 1.队列名称
             * 2.队列中的消息是否持久化到磁盘中，默认在内存中
             * 3.该队列是否能被多个消费者共享
             * 4.最后一个消费者端口连接后是否自动删除队列
             * 5.其他参数[延迟队列，死信队列]
             */
            //channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //发送消息
            String message = "hello rabbitmq";
            /* 发送消息 basicPublish
             * 1.发送到那个交换机，空字符串表示默认交换机
             * 2.路由key值，可以是队列名称
             * 3.其他参数信息
             * 4.发送消息的消息体
             */

            /*
             * 线程安全有序的一个哈希表，适用于高并发的情况下
             * 1.轻松的将序号和消息关联
             * 2.可以轻松的批量删除
             * 3.支持高并发
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms =
                    new ConcurrentSkipListMap<>();

            //消息确认成功回调函数 ack
            ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                //消息成功回调函数 - 删除确认消息编号(剩下的是未被确认的消息)
                if (multiple) {
                    //批量删除
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                    confirmed.clear();
                } else {
                    outstandingConfirms.remove(deliveryTag);
                }
            };
            //消息确认失败回调函数 nack
            ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
                System.out.println("未确认消息" + outstandingConfirms.get(deliveryTag));
            };
            channel.addConfirmListener(ackCallback, nackCallback);

            long begin = System.currentTimeMillis();
            for (int i = 1; i <= 1000; i++) {
                channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, (message + i).getBytes(StandardCharsets.UTF_8));
                //记录所有发布的消息 key:序号 value:消息体
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message + i);
            }

            long end = System.currentTimeMillis();
            System.out.println("单次确认发布耗时" + (end - begin) + "毫秒");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            RabbitMqUtils.close(connection, channel);
        }
    }
}
